package pers.vic.mq.kafka.listener;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.support.ProducerListener;


/**
 * kafkaProducer监听器，在producer配置文件中开启
 *
 */
public class KafkaProducerListener implements ProducerListener<Object, Object>{
	private static Logger logger = LoggerFactory.getLogger(KafkaProducerListener.class);
	//	 发送消息成功后调用
	public void onSuccess(String topic, Integer partition, Object key,
                          Object value, RecordMetadata recordMetadata) {
		logger.info("==========kafka发送数据成功：topic:"+topic+"，partition:"+partition+"，key:"+key+"，value:"+value+"，RecordMetadata:"+recordMetadata);
	}

//	 * 发送消息错误后调用
	public void onError(String topic, Integer partition, Object key,
                        Object value, Exception exception) {
		logger.info("==========kafka发送数据错误：topic:"+topic+"---partition:"+partition+"---key:"+key+"---value:"+value+"---:");
		logger.error(ExceptionUtils.getStackTrace(exception));
	}

	//	 方法返回值代表是否启动kafkaProducer监听器
	public boolean isInterestedInSuccess() {
		return true;
	}
}
